package com.activemq.tx;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.IdUtil;
import com.activemq.MqHelloWorld;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.util.Objects;

/**
 * '
 * 自动签收：快递放到门口算签收
 * 手动签收：亲手接到快递算签收
 * 签收主要是偏消费者
 * 事务
 */
@Slf4j
public class MqAckTx {
}

@Slf4j
class Sender {
    public static void main(String[] args) throws Exception {
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;
        try {
            ConnectionFactory factory = new ActiveMQConnectionFactory(MqHelloWorld.URL);
            connection = factory.createConnection();
            connection.start();
            //设置为手动签收
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue(MqAck.ACK_QUEUE);
            producer = session.createProducer(queue);
            for (int i = 0; i < 10; i++) {
                producer.send(session.createTextMessage("测试消息:" + IdUtil.getSnowflakeNextIdStr() + "-" + DateUtil.date().toTimeStr()));
            }
        } catch (Exception e) {
            log.error("出错", e);
        } finally {
            if (Objects.nonNull(producer)) {
                producer.close();
            }
            if (Objects.nonNull(session)) {
                session.close();
            }
            if (Objects.nonNull(connection)) {
                connection.close();
            }
        }
    }
}

@Slf4j
class TxAckReceive {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ActiveMQConnectionFactory(MqHelloWorld.URL);
        Connection connection = factory.createConnection();
        connection.start();
        //手动签收使用自动签收的代码，就会重复消费
        //事务大于签收
        //事务提交，自动签收
        //事务回滚，消息可以再次接受
        Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
        Queue queue = session.createQueue(MqAck.ACK_QUEUE);
        MessageConsumer consumer = session.createConsumer(queue);
        consumer.setMessageListener((message -> {
            if (!(message instanceof TextMessage)) {
                return;
            }
            String text;
            try {
                text = ((TextMessage) message).getText();
                //手动签收，这里签收了才行，才不会重复消费。如果写了事务和commit，那么不签收也可以
                //如果是事务，那么commit不签收也行
                //但是如果是事务，acknowledge签收不写commit也会重复消费，必须要commit
//                message.acknowledge();
            } catch (Exception e) {
                throw ExceptionUtil.wrapRuntime(e);
            }
            log.error("接收到消息：{}", text);
        }));
        System.in.read();
        consumer.close();
        session.commit();
        session.close();
        connection.close();
    }
}